package sparkStream

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe


object SparkKafka {
  def main(args: Array[String]): Unit = {
    /*spark streaming 实现kafka的消费者
    * 1） 构建sparkconf 本地运行，运行应用程序名称
    * 2） 构建sparkstreaming ---》streamingcontext，加载配置
    * 3） kafka 配置 broker ，key value，group id，消费模式
    * 4） spark 链接kafka 订阅，topic，streamingcontext
    * 5） 通过循环的形式 打印/ 处理
    * 6） 开启ssc，监控 kafka 数据
    * */
    //    1） 构建sparkconf 本地运行，运行应用程序名称
    val conf = new SparkConf().setMaster("local[*]").setAppName("hi")
    //    streamingContext 需要导入依赖
    //    spark streaming 可以进行流式处理，微批次处理，间隔 2秒
    val ssc = new StreamingContext(conf, Seconds(2))

    // spark输出红色 info信息（不报错） --> error（报错）
    ssc.sparkContext.setLogLevel("error")
    //    3） kafka 配置 broker ，key value，group id，消费模式
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.244.11:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "hael",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    //    4） spark 链接kafka 订阅，topic，streamingcontext
    //    topic name
    val topicname = Array("lol")
    val streamRdd = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topicname, kafkaParams)
    )
    streamRdd.foreachRDD(
      x => {
        if (!x.isEmpty()) {
          val line = x.map(_.value())
           line.foreach(println)
        }
      }
    )
//    6） 开启ssc，监控 kafka 数据
    ssc.start()
    ssc.awaitTermination()//kafka数据
  }
}
